package com.example.streamkafka.controller;

import com.example.streamkafka.message.Sink;
import com.example.streamkafka.message.Source;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * describe:
 *
 * @author helloworldyu
 * @date 2018/6/26
 */
@RestController
public class HelloController {
	AtomicInteger count=new AtomicInteger(0);
	@Autowired
	Source source;


	@RequestMapping("hello")
	public String hello(String info){
		int co = this.count.incrementAndGet();
		source.output().send(MessageBuilder.withPayload(info).build());
		return Integer.toString(co);
	}

	@StreamListener(Sink.INPUT)
	public void receive(Message<String> message) {
		System.out.println(message.getPayload());
	}
}